package cn.kennylee.codehub.mq.lip.integration.kafka.listener;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.kennylee.codehub.mq.lip.integration.base.AbstractMqListener;
import cn.kennylee.codehub.mq.lip.integration.base.MessageVo;
import cn.kennylee.codehub.mq.lip.integration.base.MqChannel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.slf4j.MDC;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.lang.NonNull;

import java.util.Set;

/**
 * <p> 抽象Kafka消息监听器 </p>
 * <p>Created on 2025/6/12.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractKafkaListener extends AbstractMqListener<ConsumerRecord<String, String>>
    implements MessageListener<String, String> {

    /**
     * 消费者组后缀
     */
    private static final String GROUP_SUFFIX = "-GROUP";

    /**
     * 消息主题
     */
    private String topic;

    /**
     * Kafka 无 Tag，用分区或 Header 模拟
     */
    private Set<String> partitions;

    /**
     * 消费者组
     * <p>注意组名不要重复，默认组名生成规则是，类名+环境标识，一般消费者都在1个中心的话不会重复，但如果分布不同中心，存在问题。</p>
     */
    private String consumerGroup;

    /**
     * 并发消费者数量
     */
    private int concurrency = 3;

    public AbstractKafkaListener(String topic) {
        this(topic, "");
    }

    public AbstractKafkaListener(String topic, String consumerGroup) {
        this.topic = topic;
        this.consumerGroup = StrUtil.isBlank(consumerGroup) ? generateConsumerGroup() : consumerGroup;
    }

    @Override
    public void onMessage(@NonNull ConsumerRecord<String, String> record) {
        boolean success = super.processMessage(record);
        if (!success) {
            handleFailure(record);
        }
    }

    @Override
    protected void setupTraceId(@NonNull ConsumerRecord<String, String> record) {
        // 从 Kafka Headers 获取 TraceId（需发送方设置）
        String traceId = extractTraceId(record);
        // 放入MDC
        if (StrUtil.isBlank(traceId)) {
            traceId = IdUtil.fastSimpleUUID();
            log.debug("设置 Trace ID: {}", traceId);
        } else {
            log.debug("消息已经存在 Trace ID: {}", traceId);
        }
        MDC.put(MqChannel.TRACE_ID_KEY, traceId);
    }

    /**
     * 从 Kafka 消息中提取 TraceId
     *
     * @param record Kafka 消息记录
     * @return TraceId 或 null
     */
    protected String extractTraceId(ConsumerRecord<String, String> record) {
        // 从 Kafka Headers 提取 TraceId
        Iterable<Header> traceId = record.headers().headers(MqChannel.TRACE_ID_KEY);
        if (traceId == null || !traceId.iterator().hasNext()) {
            log.trace("消息缺少 traceId. Topic: {}, Key: {}", topic, record.key());
            return null;
        }
        return traceId.iterator().next().value().toString();
    }

    protected void handleFailure(ConsumerRecord<String, String> record) {
        // Kafka 无原生重试机制，需手动实现（如发送到重试 Topic）
        log.warn("消息处理失败，准备重试. Topic: {}, Key: {}", topic, record.key());
        sendToRetryTopic(record);
    }

    /**
     * 发送消息到重试 Topic
     *
     * @param record 消息记录
     */
    protected void sendToRetryTopic(ConsumerRecord<String, String> record) {
        // 实现具体的重试逻辑，例如发送到一个专门的重试 Topic
        log.warn("消息处理失败，重试逻辑未实现. Topic: {}, Key: {}", topic, record.key());
    }

    @Override
    public String getMsgId() {
        return String.format("%s-%d-%d", topic, getMessage().partition(), getMessage().offset());
    }

    @Override
    public String getChannel() {
        return topic;
    }

    @Override
    public String getRouteKey() {
        // 用分区模拟 Tag
        return String.valueOf(getMessage().partition());
    }

    @Override
    protected MessageVo convertMessageVo(ConsumerRecord<String, String> record) {
        if (log.isDebugEnabled()) {
            log.debug("接收到消息，消息体内容: {}", record.value());
        }
        try {
            String body = record.value();
            if (!JSONUtil.isTypeJSON(body)) {
                log.warn("非JSON格式消息. Topic: {}, Key: {}", topic, record.key());
                return null;
            }
            return JSONUtil.toBean(body, MessageVo.class);
        } catch (Exception e) {
            log.error("消息转换失败. Topic: {}, Offset: {}", topic, record.offset(), e);
            return null;
        }
    }

    /**
     * 生成消费者组名称
     * <p>格式为：类名 + "-GROUP"</p>
     *
     * @return 消费者组名称
     */
    protected String generateConsumerGroup() {
        return this.getClass().getSimpleName() + GROUP_SUFFIX;
    }
}
